package org.sxp.common.utils.datax.writer;

import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Maps;
import org.sxp.common.dto.datax.writer.DataXHbaseWriter;
import org.sxp.common.dto.datax.writer.DataXHiveWriter;
import org.sxp.common.exception.BusinessException;
import org.sxp.common.utils.SpringContextUtils;
import org.sxp.modules.datasource.entity.DatabaseDO;
import org.sxp.modules.datasource.service.DatabaseService;
import org.sxp.modules.dataworks.dto.SyncDTO;

import java.util.Map;
import java.util.regex.Matcher;

/**
 * mysql writer构建类
 *
 * @author zhouhongfa@gz-yibo.com
 * @ClassName MysqlWriter
 * @Version 1.0
 * @since 2019/7/30 23:08
 */
public class HbaseWriter extends BaseWriter implements IDataxWriter {

    public static final String SPLIT_KEY = "/";
    public static final String VERSION_20XSQL = "20xsql";
    public static final String VERSION_11XSQL = "11xsql";
    public static final String VERSION_11X = "11x";

    public HbaseWriter(SyncDTO syncDTO) {
        super(syncDTO);
    }

    @Override
    public String getName() {
        return "hbase" + syncDTO.getTargetDbVersion() + "writer";
    }

    @Override
    public Map<String, Object> build() {
        beforeBuild();

        DatabaseService databaseService = SpringContextUtils.getBean(DatabaseService.class);
        DatabaseDO jobDatasource = databaseService.getById(syncDTO.getTargetDb());
        Map<String, Object> writerObj = Maps.newLinkedHashMap();
        writerObj.put("name", getName());
        Map<String, Object> parameterObj = Maps.newLinkedHashMap();

        if(syncDTO.getTargetDbVersion().equals(VERSION_20XSQL)){
            version20XSql(parameterObj, jobDatasource);
        }else if(syncDTO.getTargetDbVersion().equals(VERSION_11X)){
            version11X(parameterObj, jobDatasource);
        }else{
            throw new BusinessException("不支持的Hbase版本:"+syncDTO.getTargetDbVersion());
        }

        writerObj.put("parameter", parameterObj);
        return writerObj;
    }

    @Override
    public void backBuild() {
        DatabaseService databaseService = SpringContextUtils.getBean(DatabaseService.class);
        DatabaseDO jobDatasource = databaseService.getById(syncDTO.getTargetDb());

        String removeSpace = StrUtil.trim(syncDTO.getDataxJson());
        /*
         * 根据key和josn字符串取出特定的value :{"(.*?)"}
         */
        Matcher matcher = pattern.matcher(removeSpace);
        String value = null;
        if (matcher.find()) {
            value = matcher.group(1);
        }
        if(JSONUtil.isJson(value)){
            DataXHbaseWriter writer = JSON.parseObject(value, DataXHbaseWriter.class);
            if(ArrayUtil.isNotEmpty(writer.getParameter().getRowkeyColumn())){
                syncDTO.setRowkeyColumn(JSON.toJSONString(writer.getParameter().getRowkeyColumn()));
            }
        }
    }

    private void version20XSql(Map<String, Object> parameterObj, DatabaseDO jobDatasource) {

    }

    private void version11XSql(Map<String, Object> parameterObj, DatabaseDO jobDatasource) {

    }

    private void version11X(Map<String, Object> parameterObj, DatabaseDO jobDatasource) {
        parameterObj.put("table", syncDTO.getTargetTable());
        parameterObj.put("mode", "normal");
        parameterObj.put("column", syncDTO.getTargetField());
        parameterObj.put("encoding", syncDTO.getTargetEncoding());
        parameterObj.put("rowkeyColumn", JSON.parseArray(syncDTO.getRowkeyColumn()));
        parameterObj.put("versionColumn", JSON.parseObject(syncDTO.getVersionColumn()));

        JSONObject jsonObject = JSON.parseObject(syncDTO.getTargetHbaseConfig());
        if(jsonObject == null){
            jsonObject = new JSONObject();
        }
        jsonObject.put("hbase.zookeeper.quorum", jobDatasource.getUrl());
        parameterObj.put("hbaseConfig", jsonObject);
    }
}
